package com.asinking.sink;

import com.asinking.debezium.DebeziumObject;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Collection;
import java.util.HashMap;

public class Sink2Ck extends RichSinkFunction<DebeziumObject> {


    private String sql = null;
    private Connection connection;
    private PreparedStatement ps;
    private int size = 0;

    public Sink2Ck() {
        super();
    }

    @Override
    public void open(Configuration parameters) throws Exception {

        String driverClass = "ru.yandex.clickhouse.ClickHouseDriver";
        Class.forName(driverClass);
        String url = "jdbc:clickhouse://10.49.0.46:8123/test";
        String username = "akflow_ch";
        String password = "apple123Ab!";

        connection = DriverManager.getConnection(url, username, password);

    }

    @Override
    public void close() throws Exception {
        super.close();
        ps.close();
        connection.close();
    }

    @Override
    public void invoke(DebeziumObject debeziumObject, Context context) throws Exception {
        if (debeziumObject.isDDL()) {
            ps.execute(debeziumObject.getDdlString());
        } else {
            String rowKind = debeziumObject.getRowKind();
            if ("INSERT".equals(rowKind)) {
                insert(1, 9, debeziumObject, debeziumObject.getValues());
            } else if ("UPDATE".equals(rowKind)) {
                insert(-1, 9, debeziumObject, debeziumObject.getOldValues());
                insert(1, 9, debeziumObject, debeziumObject.getValues());
            } else {
                insert(-1, 9, debeziumObject, debeziumObject.getValues());
            }

        }
    }

    public void insert(int sign, int version, DebeziumObject debeziumObject, HashMap<String, Object> valuesHashMap) throws SQLException {

        toSql(debeziumObject, valuesHashMap);
        ps.setInt(size + 1, version);
        ps.setInt(size + 2, sign);
        System.out.println(debeziumObject);
        ps.executeUpdate();

    }


    public void toSql(DebeziumObject debeziumObject, HashMap<String, Object> valuesHashMap) throws SQLException {

        size = debeziumObject.getValues().size();
        Object[] keys = valuesHashMap.keySet().toArray();
        StringBuffer filedStr = new StringBuffer();
        StringBuffer valStr = new StringBuffer();
        for (int i = 0; i < size; i++) {
            filedStr.append(keys[i]).append(",");
            valStr.append("?").append(",");
        }
        filedStr.append("version").append(",").append("sign");

        valStr.append("?").append(",").append("?");

        sql = "INSERT INTO test_29 ( " + filedStr + " ) VALUES ( " + valStr + " )";

        ps = connection.prepareStatement(sql);

        Collection<Object> values = debeziumObject.getValues().values();
        int i = 1;
        for (Object val : values) {
            ps.setObject(i, val);
            i += 1;
        }
    }

}
